<?php abstract class Connections {
	const READ_BUF_SIZE = 1024;
	protected static $Connections = array ();
	protected static $SendBuffer = array ();

	public static function Push ($Connection) {
		if (! is_resource ($Connection)) {
			return (FALSE);
		}
		if (! @socket_set_nonblock ($Connection)) {
			$Error = socket_last_error ($Connection);
			Daemon::Warn ('M__CON_ERR_NONBLOCK', socket_strerror ($Error), $Error);
			return (FALSE);
		}
		array_push (self::$Connections, $Connection);
		return (TRUE);
	}
	public static function Down () {
		foreach (self::$Connections as $Connection) {
			@socket_shutdown ($Connection);
			@socket_close ($Connection);
		}
	}
	protected static function Read ($SID) {
		if (! array_key_exists ($SID, self::$Connections)) {
			return (FALSE);
		}
		$Result = $Buf = '';
		while ($R = @socket_recv (self::$Connections [$SID], $Buf, self::READ_BUF_SIZE, 0)) {
			$Result .= $Buf;
		}
		if ((0 === $R) and (NULL === $Buf)) { // socket is closed at remote end
			socket_shutdown (self::$Connections [$SID]);
			socket_close (self::$Connections [$SID]);
			unset (self::$Connections [$SID]);
		}
		return ($Result);
	}
	protected static function Write ($SID, $Message) {
		if (! array_key_exists ($SID, self::$Connections)) {
			return (FALSE);
		}
		while (strlen ($Message)) {
			$R = @socket_send (self::$Connections [$SID], $Message, strlen ($Message), 0);
			if ($R) {
				$Message = substr ($Message, $R);
			} else {
				return (FALSE);
			}
		}
		return (TRUE);
	}
	public static function Receive () {
		$Result = array ();
		if (count (self::$Connections)) {
			$Read = self::$Connections;
			$N = NULL;
			if (FALSE === @socket_select ($Read, $N, $N, 0)) {
				$Error = socket_last_error ();
				Daemon::Warn ('M__CON_ERR_SELECT', socket_strerror ($Error), $Error);
				return (FALSE);
			}
			foreach (array_keys ($Read) as $SID) {
				$Buf = self::Read ($SID);
				if ($Buf) {
					$Result [$SID] = $Buf;
				}
			}
		}
		return ($Result);
	}
	public static function Transmit (array $Messages) {
		// spool the messages
		foreach ($Messages as $SID => $Message) {
			if (! array_key_exists ($SID, self::$SendBuffer)) {
				self::$SendBuffer [$SID] = array ();
			}
			array_push (self::$SendBuffer [$SID], $Message);
		}
		// process the spool
		$Write = array ();
		$N = NULL;
		foreach (array_keys (self::$SendBuffer) as $SID) {
			if (array_key_exists ($SID, self::$Connections)) {
				$Write [$SID] = self::$Connections [$SID];
			} else {
				unset (self::$SendBuffer [$SID]);
			}
		}
		// send the spool
		if (count ($Write)) {
			if (FALSE === @socket_select ($N, $Write, $N, 0)) {
				$Error = socket_last_error ();
				Daemon::Warn ('M__CON_ERR_SELECT', socket_strerror ($Error), $Error);
				return (FALSE);
			}
			foreach (array_keys ($Write) as $SID) {
				foreach (self::$SendBuffer [$SID] as $MID => $Message) {
					$R = self::Write ($SID, $Message);
					if ($R) {
						unset (self::$SendBuffer [$SID][$MID]);
					}
				}
			}
		}
		return (TRUE);
	}
} ?>